

package rafthttp

import (
	"bytes"
	"context"
	"errors"
	"io"
	"runtime"
	"sync"
	"time"

	"go.etcd.io/etcd/client/pkg/v3/types"
	"go.etcd.io/etcd/pkg/v3/pbutil"
	"go.etcd.io/etcd/raft/v3"
	"go.etcd.io/etcd/raft/v3/raftpb"
	stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"

	"go.uber.org/zap"
)

const (
	connPerPipeline = 4
	// pipelineBufSize是管道缓冲区的大小，有助于保存
	// 临时网络延迟。
	// 该大小可确保当网络
	// 在良好路径中不工作不到1秒时，管道不会丢弃消息。
	pipelineBufSize = 64
)

var errStopped = errors.New("stopped")

// pipeline负责传输快照数据
type pipeline struct {
	peerID types.ID	// 该pipeline对应节点的ID。
	tr     *Transport	// 关联的ra他ttp.Transport实例。
	picker *urlPicker
	status *peerStatus
	raft   Raft	// 底层的Raft实例。
	errorc chan error
	// 当我们解析v2 API 
	followerStats *stats.FollowerStats
	msgc chan raftpb.Message	// pipeline实例从该通道中获取待发送的消息。
	// 负责同步多个goroutine结束。每个pipeline实例会启动多个后台goroutine（默认值是4个）来处理msgc通道中的消息，
	// 在pipeline.stop（）方法中必须等待这些goroutine都结束（通过wg.Wait（）方法实现），才能真正关闭该pipeline实例。
	wg    sync.WaitGroup
	stopc chan struct{}
}

// pipeline.start（）方法中会初始化上述字段，同时会启动用来发送消息的后台goroutine，
func (p *pipeline) start() {
	p.stopc = make(chan struct{})
	// 注意缓存，默认是64，主妥是为了防止瞬间网络延迟造成消息丢失
	p.msgc = make(chan raftpb.Message, pipelineBufSize)
	p.wg.Add(connPerPipeline)	// 初始化sync.WaitGroup
	for i := 0; i < connPerPipeline; i++ {	// 启动用于发送消息的gorout工口e（默认是4个）
		go p.handle()
	}

	if p.tr != nil && p.tr.Logger != nil {
		p.tr.Logger.Info(
			"started HTTP pipelining with remote peer",
			zap.String("local-member-id", p.tr.ID.String()),
			zap.String("remote-peer-id", p.peerID.String()),
		)
	}
}

func (p *pipeline) stop() {
	close(p.stopc)
	p.wg.Wait()

	if p.tr != nil && p.tr.Logger != nil {
		p.tr.Logger.Info(
			"stopped HTTP pipelining with remote peer",
			zap.String("local-member-id", p.tr.ID.String()),
			zap.String("remote-peer-id", p.peerID.String()),
		)
	}
}

// pipeline.handle（）方法中会从msgc通道中读取待发送的Message消息，然后调用pipeline.post（）方法将其发送出去，
// 发送结束之后会调用底层Raft接口的相应方法报告发送结果。
func (p *pipeline) handle() {
	defer p.wg.Done()	// handle （）方法执行完成，也就是当前这个goroutine结束

	for {
		select {
		case m := <-p.msgc:	// 获取待发送的MsgSnap类型的Message
			start := time.Now()
			err := p.post(pbutil.MustMarshal(&m))	// 将消息序列化，然后创建HTTP请求并发送出去
			end := time.Now()

			if err != nil {
				p.status.deactivate(failureType{source: pipelineMsg, action: "write"}, err.Error())

				if m.Type == raftpb.MsgApp && p.followerStats != nil {
					p.followerStats.Fail()
				}
				// 向底层的Raft状态机报告失败信息，
				p.raft.ReportUnreachable(m.To)
				if isMsgSnap(m) {
					p.raft.ReportSnapshot(m.To, raft.SnapshotFailure)
				}
				sentFailures.WithLabelValues(types.ID(m.To).String()).Inc()
				continue
			}

			p.status.activate()
			if m.Type == raftpb.MsgApp && p.followerStats != nil {
				p.followerStats.Succ(end.Sub(start))
			}
			// 向底层的Raft状态机报告发送成功的信息
			if isMsgSnap(m) {
				p.raft.ReportSnapshot(m.To, raft.SnapshotFinish)
			}
			sentBytes.WithLabelValues(types.ID(m.To).String()).Add(float64(m.Size()))
		case <-p.stopc:
			return
		}
	}
}

// pipeline. post（）方法是真正完成消息发送的地方，其中会启动一个后台goroutine监听控制发送过程及获取发送结果
func (p *pipeline) post(data []byte) (err error) {
	u := p.picker.pick()	// 获取对立在暴露的URL地址
	// 创建HTTPPOST请求
	req := createPostRequest(p.tr.Logger, u, RaftPrefix, bytes.NewBuffer(data),
		"application/protobuf", p.tr.URLs, p.tr.ID, p.tr.ClusterID)

	done := make(chan struct{}, 1)	// 主妥用于通知下面的goroutine请求是否已经发送完成
	ctx, cancel := context.WithCancel(context.Background())
	req = req.WithContext(ctx)
	go func() {	// 该goroutine主妥用于监听请求是否需妥取消
		select {
		case <-done:
			cancel()
		case <-p.stopc:	// 如采在请求的发送过程中，pipeline被关闭，则取消该请求
			waitSchedule()
			cancel()	// 取消请求
		}
	}()
	// 发送上述HTTPPOST请求，并获取到对应的响应，
	resp, err := p.tr.pipelineRt.RoundTrip(req)
	done <- struct{}{}	// 通知上述goroutine，请求已经发送完毕
	if err != nil {
		p.picker.unreachable(u)
		return err
	}
	defer resp.Body.Close()
	b, err := io.ReadAll(resp.Body)	// 读取HttpResponse.Body内容
	if err != nil {
		p.picker.unreachable(u)	// 出现异常时，则将该URL标识为不可用，再尝试其他URL地址
		return err
	}
	err = checkPostResponse(p.tr.Logger, resp, b, req, p.peerID)	// 检测响应的内容，
	if err != nil {
		p.picker.unreachable(u)
		// errMemberRemoved是一个严重错误，因为删除的成员应始终停止。因此，我们使用reportCriticalError将其报告给errorc。
		if err == errMemberRemoved {
			reportCriticalError(err, p.errorc)
		}
		return err
	}

	return nil
}

// 等待调度其他goroutine一段时间
func waitSchedule() { runtime.Gosched() }
